package org.ws.task.netty.handler;

import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.exception.RequestHandlerException;
import org.ws.task.item.ExecuteContext;
import org.ws.task.item.ItemLeaderCommit;
import org.ws.task.item.TaskInstance;
import org.ws.task.model.TaskProcessStatus;
import org.ws.task.model.protocol.CommitResult;
import org.ws.task.repository.LeaderContextRepository;
import org.ws.task.repository.TaskRepository;
import org.ws.task.model.protocol.CommitRequest;
import org.ws.task.model.protocol.CommitResponse;
import org.ws.task.model.TaskItem;
import org.ws.task.netty.RequestActionHandler;
import org.ws.task.model.protocol.base.RequestAction;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.model.protocol.base.ResponseMessage;
import org.ws.task.model.protocol.base.ResponseStatus;

import java.util.List;

public class CommitStatusHandler implements RequestActionHandler<CommitRequest, CommitResponse> {

    private static Logger log = LoggerFactory.getLogger(CommitStatusHandler.class.getName());

    @Override
    public RequestAction action() {
        return RequestAction.COMMIT_STATUS;
    }

    @Override
    public ResponseMessage<CommitResponse> handler(ChannelHandlerContext ctx, RequestMessage<CommitRequest> requestMessage) throws RequestHandlerException {
        int id = requestMessage.getId();
        ResponseMessage<CommitResponse> responseMessage = new ResponseMessage<>(id);
        CommitRequest commitRequest = requestMessage.getBodyObject();
        String taskName = commitRequest.getTaskName();
        String nodeId = commitRequest.getNodeId();
        TaskInstance taskInstance = TaskRepository.getInstance().getTaskInstance(taskName);
        if(taskInstance != null){
            ItemLeaderCommit itemLeaderCommit = taskInstance.getItemLeaderCommit();
            ExecuteContext context = LeaderContextRepository.getInstance().getContext(taskName, nodeId);
            List<TaskItem> items = commitRequest.getItems();
            CommitResponse commitResponse = new CommitResponse();
            if(CollectionUtils.isNotEmpty(items)){
                for (TaskItem item : items) {
                    CommitResult commitResult = new CommitResult();
                    commitResult.setItemId(item.getId());
                    try {
                        boolean result = itemLeaderCommit.commitStatus(item,context);
                        commitResult.setSuccess(result);
                    } catch (Exception e) {
                        log.error(e.getMessage(),e);
                        commitResult.setSuccess(false);
                        commitResult.setMessage(e.getMessage());
                    }
                    if(commitResult.getSuccess()){
                        context.commitStatus(item.getId(), TaskProcessStatus.SUCCESS);
                    }
                    else {
                        context.commitStatus(item.getId(), TaskProcessStatus.COMMIT_FAILED);
                    }
                    commitResponse.addResult(commitResult);
                }
            }

            responseMessage.setSuccess(true);
            responseMessage.setStatus(ResponseStatus.SUCCESS.getCode());
            responseMessage.setBodyObject(commitResponse);
        }
        else {
            responseMessage.setSuccess(false);
            responseMessage.setStatus(ResponseStatus.NO_HANDLER.getCode());
        }
        return responseMessage;
    }
}
